
#include "rpointd.h"

void rp_rpointd_log_init()
{
	char fname[] = "rp_rpointd_log_init";
	char msg[] = "fatal error";
	Err err;

	if (rp_log_init(config.logdir, "rpointd", config.loglv) != E_OK) exit(127);
}

void rp_rpointd_env_init()
{
	struct stat envstat;
	
	if ((env.cfgdir = getenv("RP_CFG_DIR")))
	{
		if (stat(env.cfgdir, &envstat) < 0) 
		{
			env.cfgdir = "/etc/rally/bin";
		}		
	} else {
		env.cfgdir = "/etc/rally/bin";
	}
	if (stat(env.cfgdir, &envstat) < 0) 
	{
		exit(127);
	}
	
	if ((env.bindir = getenv("RP_BIN_DIR")))
	{
		if (stat(env.bindir, &envstat) < 0) 
		{
			env.bindir = "/etc/rally/bin";
		}		
	} else {
		env.bindir = "/etc/rally/bin";
	}
	if (stat(env.bindir, &envstat) < 0) 
	{
		exit(127);
	}	
}

void rp_rpointd_config_init()
{
	/* init cfgv array  */
	config.cfgc = 8;
	config.cfgv = (KeyValue*)calloc(config.cfgc, sizeof(KeyValue));
	config.cfgv[0].key = "LogDir";
	config.cfgv[1].key = "LogLevel";
	config.cfgv[2].key = "ThreadPrestart";
	config.cfgv[3].key = "NetworkTimeOut";
	config.cfgv[4].key = "HandshakeIPAddress";
	config.cfgv[5].key = "HandshakePortNum";
	config.cfgv[6].key = "IOIPAddress";
	config.cfgv[7].key = "IOPortNum";
	
	/*	read from config file   */
	if (rp_config_init(env.cfgdir, "rpointd", config.cfgv, config.cfgc) == E_FATAL)  exit(127);
	
	/*	convert from string to real type   */
	config.logdir = strdup(config.cfgv[0].value);
	config.num_thread_prestart = atoi(config.cfgv[2].value);
	config.tv_io_timeout.tv_sec = atol(config.cfgv[3].value); 
	int i;
	for (i=0; i<5; i++)
	{
		if (strcmp(config.cfgv[1].value, log_level_array[i]) == 0)
		{
			config.loglv = (Log_Level)i;
			break;
		}
	}

	config.addr_hsk.sin_family = AF_INET;
	config.addr_hsk.sin_port = htons((uint16_t)atoi(config.cfgv[5].value));
	config.addr_hsk.sin_addr.s_addr = inet_addr(config.cfgv[4].value);
	memset(&config.addr_hsk.sin_zero, 0, 8);

	config.addr_io.sin_family = AF_INET;
	config.addr_io.sin_addr.s_addr = inet_addr(config.cfgv[6].value);
	config.addr_io.sin_port = htons((uint16_t)atoi(config.cfgv[7].value));
	memset(&(config.addr_io.sin_zero), 0, 8);

	/*	free mem  */
	for (i=0; i<config.cfgc; i++)
	{
		free(config.cfgv[i].value);
	}
	free(config.cfgv);
	return;
}

void rp_rpointd_global_init()
{
	char fname[] = "rp_rpointd_global_init";
	char msg[] = "fatal error";
	Err err;
	FLOG_IN(fname);
	
	global.fd_stdin = STDIN_FILENO;
	if (rp_mutex_init(&(global.mutex_code)) != E_OK)
	{
		rp_log(LOG_DEBUG, fname, "Cannot initialize the code mutex ...");
	}
	if (rp_mutex_init(&(global.mutex_worker)) != E_OK)
	{
		rp_log(LOG_DEBUG, fname, "Cannot initialize the worker mutex ...");
	}	
	err = rp_sem_init(&(global.sem_worker), SEM_PRIVATE, config.num_thread_prestart);		// TODO: how to reset the sem value after one client finishes?

	global.status_worker = (Status*)calloc(config.num_thread_prestart, sizeof(Status));
	Status *status = global.status_worker;
	int i;
	for (i=0; i<config.num_thread_prestart; i++)
	{
		if (rp_mutex_init(&(status+i)->mutex) != E_OK)
		{
			rp_log(LOG_DEBUG, fname, "Cannot initialize %dth worker status mutex ...", i);
		}	
	}

	FLOG_OUT(fname);
	return;
}

Err rp_rpointd_sq_init(SessionQueue*sq)
{
	char fname[] = "rp_rpointd_sq_init";
	char msg[] = "fatal error";
	Err err;
	FLOG_IN(fname);
	
	err = rp_mutex_init(&(sq->mutex));
	err = rp_mutex_lock(&(sq->mutex));
	sq->head = NULL;
	sq->tail = NULL;
	err = rp_mutex_unlock(&(sq->mutex));

	FLOG_OUT(fname);
	return err;
}

Err rp_rpointd_sq_finalize(SessionQueue *sq)
{
	char fname[] = "rp_rpointd_sq_finalize";
	char msg[] = "fatal error";
	Err err;
	FLOG_IN(fname);
	
	err = rp_mutex_lock(&(sq->mutex));
	sq->head = NULL;
	sq->tail = NULL;
	err = rp_mutex_unlock(&(sq->mutex));	
	err = rp_mutex_destroy(&(sq->mutex));
	
	FLOG_OUT(fname);
	return err;
}

Err rp_rpointd_sq_push(SessionQueue *sq, Session *session, int size)
{
	char fname[] = "rp_rpointd_sq_push";
	char msg[] = "fatal error";
	Err err;
	Session *temp;
	FLOG_IN(fname);
	
	temp = (Session*)malloc(size);
	memcpy(temp, session, size);
	
	err = rp_mutex_lock(&(sq->mutex));
	if (sq->head == NULL)
	{
		sq->head = temp;
		sq->tail = temp;
		request->next = NULL;
		request->prev = NULL;
	} else {
		sq->tail->next = temp;
		temp->prev = sq->tail;
		temp->next = NULL;
		sq->tail = temp;
	}
	err = rp_mutex_unlock(&(sq->mutex));	
	
	FLOG_OUT(fname);
	return err;
}

Session *rp_rpointd_sq_get(Session *sq)
{
	char fname[] = "rp_rpointd_sq_get";
	char msg[] = "fatal error";
	Err err;
	Session *session;
	
	err = rp_mutex_lock(&(sq->mutex));
	if (sq->head == NULL)
	{
		session = NULL;
	} else if (sq->head == sq->tail) 
	{
		session = sq->head;
		sq->head = NULL;
		sq->tail = NULL;
	} else {
		session = sq->head;
		sq->head = sq->head->next;
		sq->head->prev = NULL;
		session->next = NULL;
	}
	
	err = rp_mutex_unlock(&(sq->mutex));	
	
	return session;
}

